<?php

namespace lib\tools\amqp;

use lib\tools\config\MqConfig;
use lib\tools\logs\RuntimeLog;
use lib\tools\singleton\Single;
use modules\tracing\zipkin\ZipkinTracing;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

/**
 * Class Rabbitmq
 * @method static AMQPStreamConnection getInstance()
 * @package lib\tools\amqp
 */
class Rabbitmq extends Single implements \constant\api\Code
{

    const SECTION = 'rabbitmq';
    const EXCHANGE = 'ex.db.async';
    const QUEUE_NAME = 'queue.db.async';
    const SEND_ROUTING_KEY = 'routing.db.test';
    const CONSUME_ROUTING_KEY = '#';
    protected static $client_id = '';


    public static function instance()
    {



        // TODO: Implement instance() method.
        $config = MqConfig::getInstance(static::SECTION);



        $connection = new AMQPStreamConnection(
            $config->host,
            $config->port,
            $config->user,
            $config->passwd
        );


        return $connection;
    }

    public static function send($routing_key = null, $queue = null, $message = '')
    {
        $connection = self::getInstance();
        $channel = $connection->channel();

        $exchange = static::EXCHANGE;
        $queue = static::QUEUE_NAME;
        $routing_key = static::SEND_ROUTING_KEY;

        // 创建一个type="direct" 、持久化的、非自动删除的交换器
        $channel->exchange_declare($exchange, 'topic');
//    $channel->exchange_declare($exchange2, 'topic');//Declares exchange 创建一个交换机
//    $channel->exchange_delete($exchange); //Deletes an exchange 删除一个交换机
//    $channel->exchange_bind($exchange2, $exchange); //Binds dest exchange to source exchange 交换机的绑定
        //创建一个持久化、非排他的、非自动删除的队列
        $channel->queue_declare($queue, false, false, false, false);

        //将交换器与队列通过路由键绑定
        $channel->queue_bind($queue, $exchange, $routing_key);

        $msg = new AMQPMessage($message);

        $channel->basic_publish($msg, $exchange, $routing_key);

        $channel->close();

        $connection->close();
    }

    public static function consume($client_id = null, $routing_key = null, $queue = null)
    {
        if (!is_null($client_id)) self::$client_id = $client_id;
        if (is_null($queue)) $queue = static::QUEUE_NAME;
        if (is_null($routing_key)) $routing_key = static::CONSUME_ROUTING_KEY;

        try {

            $connection = self::instance();
            $channel = $connection->channel();
            $exchange = static::EXCHANGE;

            // 创建一个type="direct" 、持久化的、非自动删除的交换器
            $channel->exchange_declare($exchange, 'topic', false, true, false);
            //创建一个持久化、非排他的、非自动删除的队列
            $channel->queue_declare($queue, false, false, false, false);
            //将交换器与队列通过路由键绑定
            $channel->queue_bind($queue, $exchange, $routing_key);

            echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";

            $callback = [static::class, 'callback'];

            $channel->basic_consume($queue, '', false, false, false, false, $callback);

            while(count($channel->callbacks)) {
                $channel->wait();
            }

            $channel->close();
            $connection->close();
        } catch (\Exception $e) {
            //todo 日志
            RuntimeLog::error('Rabbitmq consume', compact('e'));
            if (isset($channel)) $channel->close();
            if (isset($connection)) $connection->close();
        }
    }

    public static function callback(AMQPMessage $msg)
    {
        $delivery_info = $msg->delivery_info;
        $delivery_tag = $delivery_info['delivery_tag'];
        /**
         * @var \entity\maxwell\MaxwellResponseEntity $body
         */
        $body = json_decode($msg->body);

        /**
         * @var \PhpAmqpLib\Channel\AMQPChannel $channel
         */
        $channel = $delivery_info['channel'];

        RuntimeLog::info('Rabbitmq callback', compact('delivery_tag','body'));

        $channel->basic_ack($delivery_tag);

    }

}


